/**
 * 
 */

package com.wutianyi.nio.interfaces.impl;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Callable;

import com.wutianyi.nio.interfaces.BufferFactory;
import com.wutianyi.nio.interfaces.ChannelFacade;
import com.wutianyi.nio.interfaces.InputHandler;
import com.wutianyi.nio.interfaces.InputQueue;
import com.wutianyi.nio.interfaces.OutputQueue;

/**
 * @author wutianyi
 * 
 */
public class HandlerAdapter implements Callable<HandlerAdapter>, ChannelFacade {

	private final NioDispatcher	dispatcher;
	private final InputQueue	inputQueue;
	private final OutputQueue	outputQueue;
	private final Object		stateChangeLock	= new Object();
	private InputHandler		clientHandler;
	private SelectionKey		key				= null;
	private SelectableChannel	channel			= null;
	private volatile int		interestOps		= 0;
	private int					readyOps		= 0;
	private boolean				shuttingDown	= false;
	private volatile boolean	running			= false;
	private volatile boolean	dead			= false;

	public HandlerAdapter(InputHandler clientHandler, NioDispatcher dispatcher,
			BufferFactory bufferFactory) {

		this.dispatcher = dispatcher;
		this.clientHandler = clientHandler;
		inputQueue = new InputQueueImpl(bufferFactory);
		outputQueue = new OutputQueueImpl(bufferFactory, this);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.concurrent.Callable#call()
	 */
	@Override
	public HandlerAdapter call() throws Exception {

		try {
			drainOutput();
			fillInput();
			ByteBuffer message;
			while((message = clientHandler.nextMessage(this)) != null) {
				clientHandler.handleInput(message, this);
			}
		}finally {
			synchronized(stateChangeLock) {
				running = false;
			}
		}
		return this;
	}

	void prepareToRun() {

		synchronized(stateChangeLock) {
			interestOps = key.interestOps();
			readyOps = key.readyOps();
			running = true;
		}
	}

	void setKey(SelectionKey key) {

		this.key = key;
		channel = key.channel();
		interestOps = key.interestOps();
	}

	SelectionKey key() {

		return key;
	}

	boolean isDead() {

		return dead;
	}

	void die() {

		dead = true;
	}

	void registering() {

		clientHandler.starting(this);
	}

	void registered() {

		clientHandler.started(this);
	}

	void unregistering() {

		clientHandler.stopping(this);
	}

	void unregistered() {

		clientHandler.stopped(this);
	}

	private void enableWriteSelection() {

		modifyInterestOps(SelectionKey.OP_WRITE, 0);
	}

	private void disableWriteSelection() {

		modifyInterestOps(0, SelectionKey.OP_WRITE);
	}

	private void disableReadSelection() {

		modifyInterestOps(0, SelectionKey.OP_READ);
	}

	/**
	 * @throws IOException
	 * 
	 */
	private void fillInput() throws IOException {

		if(shuttingDown)
			return;
		int rc = inputQueue.fillFrom((ByteChannel)channel);
		if(rc == -1) {
			disableReadSelection();
			if(channel instanceof SocketChannel) {
				SocketChannel sc = (SocketChannel)channel;
				if(sc.socket().isConnected()) {
					sc.socket().shutdownInput();
				}
			}
		}
		shuttingDown = true;
		clientHandler.stopping(this);
		enableWriteSelection();
	}

	/**
	 * @throws IOException
	 * 
	 */
	private void drainOutput() throws IOException {

		if(((readyOps & SelectionKey.OP_WRITE) != 0) && (!outputQueue.isEmpty())) {
			outputQueue.drainTo((ByteChannel)channel);
		}
		if(outputQueue.isEmpty()) {
			disableWriteSelection();
			if(shuttingDown) {
				channel.close();
				clientHandler.stopped(this);
			}

		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.nio.interfaces.ChannelFacade#getInterestOps()
	 */
	@Override
	public int getInterestOps() {

		return interestOps;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.nio.interfaces.ChannelFacade#inputQueue()
	 */
	@Override
	public InputQueue inputQueue() {

		return inputQueue;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.nio.interfaces.ChannelFacade#modifyInterestOps(int, int)
	 */
	@Override
	public void modifyInterestOps(int opsToSet, int opsToReset) {

		synchronized(stateChangeLock) {
			interestOps = (interestOps | opsToSet) & (~opsToReset);
			if(!running) {
				dispatcher.enqueueStatusChange(this);
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.nio.interfaces.ChannelFacade#outputQueue()
	 */
	@Override
	public OutputQueue outputQueue() {

		return outputQueue;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.wutianyi.nio.interfaces.ChannelFacade#setHandler(com.wutianyi.nio.interfaces.InputHandler
	 * )
	 */
	@Override
	public void setHandler(InputHandler handler) {

		this.clientHandler = handler;

	}

}
